"""
Case Type   : GMS_LOB
Case Name   : GMS_LOB write、append、read、getlength、isopen。用户权限测试
Create At   : 2025/5/26
Owner       : cting66
Description :
Expect      :
History     :
"""

import os
import unittest

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class CompatibilityTest01(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.common = Common()
        self.sh_primary = CommonSH('PrimaryDbAmode')
        self.user_node = Node('PrimaryDbAmode')
        self.constant = Constant()
        self.connect_info = f'-U user1 -W Aa123456'

    def test_gms_lob(self):
        self.log.info(f'----{os.path.basename(__file__)} start----')
        text = '--GMS_LOB write、append、read、getlength、isopen。用户权限测试--'
        self.log.info(text)

        sql_cmd = f"create or replace function cast_to_raw(strdata varchar2) returns raw " \
                  f"as 'select encode(cast(\$1 as bytea), ''hex'')::raw;'LANGUAGE SQL; "
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, sql_res, '执行失败')

        sql_cmd = f'''drop extension gms_lob;
                        create extension gms_lob;'''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn(self.constant.create_extension_success, sql_res, '执行失败')

        sql_cmd = f'''drop user if exists user1 cascade;
                                create user user1 password 'Aa123456';
                           '''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, sql_res, '执行失败')

        sql_cmd = f"create table tb_1(id int,col1 blob,col2 clob); \
                    insert into tb_1 values(1,cast_to_raw('1234567890'),'中文数据内容长度测试');"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, sql_type=self.connect_info)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, sql_res, '执行失败')

        sql_cmd = f'''declare
lob1 blob;
amount int := 1;
off_set int := 6;
buf varchar(100);
lob2 blob :=cast_to_raw('abc');
begin
select col1 into lob1 from tb_1 where id = 1;
gms_lob.open(lob1,gms_lob.lob_readwrite);
raise info 'INFO: lob1 isopen: %', gms_lob.isopen(lob1);
gms_lob.read(lob1,amount,off_set,buf);
raise info 'INFO: buf: %', to_char(buf);
raise info 'INFO: buf length: %', gms_lob.getlength(buf);
gms_lob.write(lob1,amount,off_set,lob2);
gms_lob.append(lob1,lob2);
raise info 'INFO: lob1: %', to_char(lob1);
end;'''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, sql_type=self.connect_info)
        self.log.info(sql_res)
        self.assertIn('lob1 isopen: 1', sql_res, '用户权限测试失败')
        self.assertIn('buf: 36', sql_res, '用户权限测试失败')
        self.assertIn('buf length: 2', sql_res, '用户权限测试失败')
        self.assertIn('lob1: 12345a7890abc', sql_res, '用户权限测试失败')

        sql_cmd = f'''declare
lob1 clob;
amount int := 10;
off_set int := 7;
buf varchar2(20);
lob2 clob :='测试abc';
begin
select col2 into lob1 from tb_1 where id = 1;
gms_lob.open(lob1,gms_lob.lob_readwrite);
raise info 'INFO: lob1 isopen: %', gms_lob.isopen(lob1);
gms_lob.read(lob1,amount,off_set,buf);
raise info 'INFO: buf: %', to_char(buf);
raise info 'INFO: buf length: %', gms_lob.getlength(buf);
gms_lob.write(lob1,amount,off_set,lob2);
gms_lob.append(lob1,lob2);
raise info 'INFO: lob1: %', to_char(lob1);
end;'''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, sql_type=self.connect_info)
        self.log.info(sql_res)
        self.assertIn('lob1 isopen: 1', sql_res, '用户权限测试失败')
        self.assertIn('buf: 长度测试', sql_res, '用户权限测试失败')
        self.assertIn('buf length: 4', sql_res, '用户权限测试失败')
        self.assertIn('lob1: 中文数据内容测试ab测试abc', sql_res, '用户权限测试失败')

    def tearDown(self):
        self.log.info('-----step4:清理环境; expect:成功-----')
        sql_cmd = f"drop user if exists user1 cascade;"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('DROP ROLE', sql_res, '用户删除失败')
        sql_cmd = f"drop function cast_to_raw(strdata varchar2);"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('DROP FUNCTION', sql_res, '函数删除失败')
        sql_cmd = f"drop extension gms_lob;"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('DROP EXTENSION', sql_res, '插件删除失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')


